package anet

import (
	"context"
	"fmt"
	"sync/atomic"
	"time"

	"gitee.com/AlexWoo/arpc"
	"gitee.com/AlexWoo/arpc/alog"
	"gitee.com/AlexWoo/arpc/trace"
	"google.golang.org/protobuf/proto"
)

var (
	sPool = arpc.NewPool(func() any { return &session{} })
)

func getSession() *session {
	s := sPool.Get().(*session)
	alog.LogDebug(anetLctx, "get session %p", s.name)

	return s
}

func putSession(s *session) {
	delTimer(s.ti)

	putRingBuffer(s)

	for _, msg := range s.sendReq {
		putMessage(msg)
	}

	for _, msg := range s.recvReq {
		putMessage(msg)
	}

	alog.LogDebug(s.lctx, "put session in pool")

	sPool.Put(s)
}

type sessState uint8

const (
	sessionNormal sessState = iota
	connClosed
	sessionClosing
	sessionClosed
)

var sessionStateEnum = []string{
	"sessionNormal",
	"connClosed",
	"sessionClosing",
	"sessionClosed",
}

func (ss sessState) String() string {
	if ss > sessionClosed {
		return "unknown"
	}

	return sessionStateEnum[ss]
}

const (
	maxTimeoutCount = 3
)

type session struct {
	bn *Net

	name      string
	appid     string
	channelid string
	userid    string

	conn Conn
	disp Dispatcher

	state sessState
	addr  string

	updateTimeout uint8
	lastRecvTime  time.Time
	sTiInterval   time.Duration
	sRespTimeout  time.Duration
	ti            *timer

	sendReq map[uint64]*message
	recvReq map[uint64]*message
	sendbuf *RingBuffer

	connpref string
	connsuff string
	lctx     *alog.LogCtx

	span *trace.Span
}

func createSession(name, appid, channelid, userid, traceId string, bn *Net, c Conn, addr string) *session {
	s := getSession()

	s.bn = bn

	s.name = name
	s.appid = appid
	s.channelid = channelid
	s.userid = userid

	s.updateConn(c)
	s.disp = bn.disp

	s.updateTimeout = 0
	s.lastRecvTime = time.Now()

	s.addr = addr
	s.state = sessionNormal

	if s.isClient() {
		s.sTiInterval = time.Duration(bn.conf.SessionTimeoutSec) * time.Second / maxTimeoutCount
	} else {
		s.sTiInterval = time.Duration(bn.conf.SessionTimeoutSec) * time.Second
	}
	s.sRespTimeout = time.Duration(bn.conf.ResponseTimeoutSec) * time.Second
	s.ti = addTimer(s.sTiInterval, s.bn.event, getEvent(s.name, eventType_sessTimeout, 0))

	s.sendReq = map[uint64]*message{}
	s.recvReq = map[uint64]*message{}
	s.sendbuf = getRingBuffer(bn.conf.BufferSize)

	s.lctx = &alog.LogCtx{
		Prefix:   s.Prefix,
		Suffix:   s.Suffix,
		LogLevel: alog.Level[s.bn.conf.Log.LogLevel],
	}

	if s.isClient() {
		s.span = trace.StartSpanServer(context.Background(), s.name, traceId)
	} else {
		s.span = trace.StartSpanClient(context.Background(), s.name, traceId)
	}
	s.span.SetAttributes(anetAttributeFromSession(s)...)

	return s
}

func (s *session) isClient() bool {
	return s.addr != ""
}

func (s *session) serverAddr() string {
	return s.addr
}

func (s *session) recvMsg(recvm *message) {
	// update session state
	s.updateConn(recvm.c)
	s.lastRecvTime = time.Now()
	s.updateTimeout = 0

	if s.state == connClosed {
		s.state = sessionNormal
	}

	save := false

	defer func() {
		if !save {
			putMessage(recvm)
		}
	}()

	m := recvm.m

	alog.LogDebug(s.lctx, "Recv msg %p[%s]", m, m.Str())

	s.span.AddEvent("receive", anetAttributeFromanetMsg(recvm.m, s)...)

	if m.Code == 0 { // Request
		if s.state >= sessionClosing {
			alog.LogError(s.lctx, "Recv request[%s], but session state is %d", m.Str(), s.state)
			return
		}

		if msg := s.recvReq[m.SerialId]; msg != nil { // retransmission
			alog.LogError(s.lctx, "Recv request[%s], but serialId %d exist", m.Str(), m.SerialId)
			return
		}

		s.recvReq[m.SerialId] = recvm
		save = true
		recvm.setRecvTimer(s, s.sRespTimeout)

		if m.Type == MsgType_Update { // send update response
			resp := Response(m, OK, nil)
			sendm := getMessage("", recvm.c, resp)
			s.sendMsg(sendm)

			return // update will not pass to service layer
		}

		if m.Type == MsgType_Leave { // send leave response
			resp := Response(m, OK, nil)
			sendm := getMessage("", recvm.c, resp)
			s.sendMsg(sendm)
		}
	} else { // Response
		sendm := s.sendReq[m.SerialId]
		if sendm == nil {
			alog.LogError(s.lctx, "Recv response[%s], but request not found", m.Str())
			return
		}

		req := sendm.m

		// check Type
		if req.Type != m.Type {
			alog.LogError(s.lctx, "Recv response[%s], but Type not match", recvm.m.Str())
			return
		}

		delete(s.sendReq, m.SerialId)
		putMessage(sendm)

		if req.Type == MsgType_Update { // receive Update response

			if m.Code == SessionNotExist {
				alog.LogError(s.lctx, "Recv Update response[%s], Session Not Exit", recvm.m.Str())
				s.state = sessionClosed
			}

			return // update response will not pass to service layer
		}

		if req.Type == MsgType_Leave { // receive Leave response
			s.state = sessionClosed
			return // leave response will not pass to service layer
		}
	}

	// send to application
	alog.LogDebug(s.lctx, "Dispatch msg %p[%s]", recvm, recvm.m.Str())
	s.disp.Dispatch(m)
}

func (s *session) send(sendm *message) bool {
	if sendm != nil {
		if head := s.sendbuf.Put(sendm); head != nil { // sendbuffer if full
			m := head.m

			alog.LogInfo(s.lctx, "Send message[%s], buf session buffer is full, pop head %s", sendm.m.Str(), m.Str())

			if m.Code == 0 {
				delete(s.sendReq, m.SerialId)
			}
			putMessage(head)

			bts, _ := proto.Marshal(m)
			if err := s.conn.Send(bts); err != nil {
				alog.LogError(s.lctx, "Send buffer head[%s], connnection is broken, discard", m.Str())
				s.state = connClosed
				if m.Code == 0 && !(m.Type == MsgType_Update || m.Type == MsgType_Leave) {
					resp := Response(m, RequestDiscard, nil)
					s.bn.disp.Dispatch(resp)
				}
				return false
			} else {
				s.span.AddEvent("send", anetAttributeFromanetMsg(m, s)...)
			}
		}
	}

	if s.state == connClosed {
		alog.LogError(s.lctx, "Send Msg [%s], But Conn Closed", sendm.m.Str())
		return false
	}

	// send msg in sendbuffer
	for !s.sendbuf.empty() {
		head := s.sendbuf.Head()
		bts, _ := proto.Marshal(head.m)
		if err := s.conn.Send(bts); err != nil {
			alog.LogInfo(s.lctx, "Send buffer head[%s], connection is broken", head.m.Str())
			s.state = connClosed
			return false
		} else {
			s.span.AddEvent("send", anetAttributeFromanetMsg(head.m, s)...)
		}

		// Pop sendbuf head
		s.sendbuf.PopHead()

		if head.m.Code != 0 {
			putMessage(head)
		}
	}

	return true
}

func (s *session) sendMsg(sendm *message) {
	sendm.c = s.conn

	m := sendm.m
	m.TraceId = s.span.TraceHeader()

	alog.LogDebug(s.lctx, "Send msg %p[%s]", m, m.Str())

	if m.Code == 0 { // Request
		if s.state >= sessionClosing {
			alog.LogError(s.lctx, "Send request[%s], but session state is %d", m.Str(), s.state)
			putMessage(sendm)
			return
		}

		if msg := s.sendReq[m.SerialId]; msg != nil { // retransmission
			alog.LogError(s.lctx, "Send request[%s], but serialId %d exist", m.Str(), m.SerialId)
			putMessage(sendm)
			return
		}

		s.sendReq[m.SerialId] = sendm
		sendm.setSendTimer(s, s.sRespTimeout)

		if m.Type == MsgType_Leave {
			s.state = sessionClosing
		}
	} else { // Response
		recvm := s.recvReq[m.SerialId]
		if recvm == nil {
			alog.LogError(s.lctx, "Send response[%s], but has no compatiable request", m.Str())
			putMessage(sendm)
			return
		}

		req := recvm.m

		// check Type
		if req.Type != m.Type {
			alog.LogError(s.lctx, "Send response[%s], but Type not match", sendm.m.Str())
			putMessage(sendm)
			return
		}

		delete(s.recvReq, m.SerialId)
		putMessage(recvm)

		if m.Type == MsgType_Leave {
			s.state = sessionClosed
		}
	}

	s.send(sendm)
}

func (s *session) sessTimeout() {
	alog.LogDebug(s.lctx, "session timeout")

	if s.isClient() {
		m := Request(MsgType_Update, s.appid, s.channelid, s.userid, "", "", nil)
		m.SerialId = atomic.AddUint64(&s.bn.serialId, 1)
		msg := getMessage("", s.conn, m)
		s.sendMsg(msg)
		s.ti = addTimer(s.sTiInterval, s.bn.event, getEvent(s.name, eventType_sessTimeout, 0))
	} else {
		newInterval := s.sTiInterval - time.Since(s.lastRecvTime)
		if newInterval > 0 {
			s.ti = addTimer(newInterval, s.bn.event, getEvent(s.name, eventType_sessTimeout, 0))
		} else {
			s.state = sessionClosed
			s.ti = nil // avoid to del timer twice
		}
	}
}

func (s *session) recvTimeout(ev *event) {
	msg := s.recvReq[ev.sid]
	if msg == nil {
		alog.LogInfo(s.lctx, "Receive msg timeout, but received request(%d) cannot found", ev.sid)
		return
	}

	alog.LogError(s.lctx, "Receive msg[%s] timeout", msg.m.Str())

	delete(s.recvReq, ev.sid)
	msg.ti = nil // avoid to del timer twice
	putMessage(msg)
}

func (s *session) sendTimeout(ev *event) {
	msg := s.sendReq[ev.sid]
	if msg == nil {
		alog.LogInfo(s.lctx, "Send msg timeout, but send request(%d) cannot found", ev.sid)
		return
	}

	m := msg.m

	if m.Type == MsgType_Leave {
		s.state = sessionClosed
	}

	if m.Type == MsgType_Update {
		s.updateTimeout++
	}

	if s.updateTimeout >= maxTimeoutCount {
		s.state = sessionClosed
	}

	if !(m.Type == MsgType_Update || m.Type == MsgType_Leave) {
		resp := Response(m, RequestTimeout, nil)
		s.bn.disp.Dispatch(resp)
	}

	alog.LogError(s.lctx, "Send msg[%s] timeout", m.Str())

	delete(s.sendReq, ev.sid)
	msg.ti = nil // avoid to del timer twice
	putMessage(msg)
}

func (s *session) close() {
	s.disp.Dispatch(ClosedMessage(s.appid, s.channelid, s.userid))

	s.span.End()

	putSession(s)
}

func (s *session) connect() {
	s.send(nil)
}

func (s *session) updateConn(c Conn) {
	s.conn = c
	s.connpref = c.Prefix()
	s.connsuff = c.Suffix()
}

func (s *session) Prefix() string {
	if s.connpref == "" {
		return fmt.Sprintf("session %p[%s]", s, s.name)
	} else {
		return fmt.Sprintf("%s session %p[%s]", s.connpref, s, s.name)
	}
}

func (s *session) Suffix() string {
	if s.connsuff == "" {
		return fmt.Sprintf("state: %s, update timeout: %d, recvReq: %v, sendReq: %v, sendbuf Size: %d, traceid: %s, spanid: %s",
			s.state, s.updateTimeout, s.recvReq, s.sendReq, s.sendbuf.BufferSize(), s.span.TraceId(), s.span.SpanId())
	} else {
		return fmt.Sprintf("%s state: %s, update timeout: %d, recvReq: %v, sendReq: %v, sendbuf Size: %d, traceid: %s, spanid: %s",
			s.connsuff, s.state, s.updateTimeout, s.recvReq, s.sendReq, s.sendbuf.BufferSize(), s.span.TraceId(), s.span.SpanId())
	}
}
